package com.ryenlii.datatrans.core.kafka;

import com.ryenlii.datatrans.core.constant.TopicConst;
import com.ryenlii.datatrans.service.impl.ProducerServiceImpl;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.TopicPartition;


// @Component
public class KafkaClient2Demo {
    private static final Logger logger = LoggerFactory.getLogger(ProducerServiceImpl.class);

    @KafkaListener(id = TopicConst.GROUP_ID + 0, containerFactory = "kafkaListenerContainerFactory",
            // 消费指定分区
            topicPartitions = {@TopicPartition(topic = TopicConst.PLATFORM_SUBSCRIBE_CITY_DATA, partitions = {"0,1,2"})})
    public void onMessage(ConsumerRecord<String, Object> message) {
        logger.info("onMessage topic:{},partition:{}，消费消息：{}", message.topic(), message.partition(), message.value());
    }

    @KafkaListener(id = TopicConst.GROUP_ID + 1, containerFactory = "kafkaListenerContainerFactory",
            // 消费指定分区
            topicPartitions = {@TopicPartition(topic = TopicConst.PLATFORM_SUBSCRIBE_CITY_DATA, partitions = {"3,4"})})
    public void onMessage2(ConsumerRecord<String, Object> message) {
        logger.info("onMessage2 topic:{},partition:{}，消费消息：{}", message.topic(), message.partition(), message.value());
    }
}